/*
 * @(#)ExecSqlSink.java   1.0  2021年11月8日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.sink;

import java.util.Date;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.dao.RecoverableDataAccessException;
import org.springframework.stereotype.Component;

import com.swift.dao.db.datasource.HandlerDataSource;
import com.swift.dao.db.springjdbc.Jdbc;
import com.swift.dts.dao.ErrorSqlTableDao;
import com.swift.dts.dao.LogEventTableDao;
import com.swift.dts.dao.OfflineSqlTableDao;
import com.swift.dts.dao.model.ErrorSqlTable;
import com.swift.dts.dao.model.ErrorSqlTableKey;
import com.swift.dts.dao.model.LogEventTable;
import com.swift.dts.dao.model.OfflineSqlTable;
import com.swift.dts.dao.model.SynObjectKey;
import com.swift.dts.loadconfig.model.DtsDataBase;
import com.swift.dts.model.DtsBinlogMessage;
import com.swift.dts.model.SqlObject;
import com.swift.dts.offline.OfflineLifeCycle;
import com.swift.util.bean.BeanUtil;
import com.swift.util.date.DateUtil;
import com.swift.util.type.TypeUtil;

/**
 * 添加说明
 * 
 * @author zhengjiajin
 * @version 1.0 2021年11月8日
 */
@Component
public class ExecSqlSink {

    private static final Logger log = LoggerFactory.getLogger(ExecSqlSink.class);
    @Autowired
    private Jdbc jdbc;

    @Autowired
    private LogEventTableDao logEventTableDao;

    @Autowired
    private ErrorSqlTableDao errorSqlDao;

    @Autowired
    private OfflineLifeCycle offlineLifeCycle;

    @Autowired
    private OfflineSqlTableDao offlineSqlTableDao;
    
    //@Autowired
    //private OfflineStateInfoDao offlineStateInfoDao;

    public void execErrorSql(Long sqlId) {
        if (sqlId == null) return;
        ErrorSqlTable errorSqlTable = errorSqlDao.get(sqlId);
        if (errorSqlTable == null) return;
        execErrorSql(errorSqlTable);
        new ThreadErrorNextSql(errorSqlTable).run();
    }

    private class ThreadErrorNextSql extends Thread {

        private ErrorSqlTable lastSql;

        private ThreadErrorNextSql(ErrorSqlTable lastSql) {
            this.lastSql = lastSql;
            super.setDaemon(true);
        }

        public void run() {
            ErrorSqlTableKey key = new ErrorSqlTableKey();
            key.setDataSource(lastSql.getDataSource());
            key.setDestination(lastSql.getDestination());
            key.setDbName(lastSql.getDbName());
            key.setTableName(lastSql.getTableName());
            List<ErrorSqlTable> list = errorSqlDao.list(key);
            if (TypeUtil.isNull(list)) return;
            list.forEach(errorSql -> {
                execErrorSql(errorSql);
            });
        }

    }

    private void execErrorSql(ErrorSqlTable errorSqlTable) {
        HandlerDataSource.putDataSource(errorSqlTable.getDestination());
        log.info("执行异常SQL:" + errorSqlTable.getSqlStr());
        jdbc.execute(errorSqlTable.getSqlStr());
        HandlerDataSource.clear();
        errorSqlDao.delete(errorSqlTable.getSqlId());
    }

    private void execRunSql(DtsDataBase todb, DtsBinlogMessage message, SqlObject sqlObject) {
        HandlerDataSource.putDataSource(todb.getKey());
        try {
            if (offlineFilter(todb.getKey(), message, sqlObject)) return;
            if(todb.getIsRing()) {
                LogEventTable eventTable = new LogEventTable();
                eventTable.setCreateTime(new Date());
                eventTable.setSqlKey(sqlObject.getUnionKey());
                eventTable.setSqlStr(sqlObject.getSql());
                //eventTable.setState(0);
                //Long sqlId = logEventTableDao.save(eventTable);
                eventTable.setState(1);
                eventTable.setUpdateTime(new Date());
                logEventTableDao.save(eventTable);
            }
            log.info("执行SQL:" + sqlObject.getSql());
            jdbc.execute(sqlObject.getSql());
            /*eventTable.setState(1);
            eventTable.setUpdateTime(new Date());
            logEventTableDao.updateSuccess(sqlId);*/
        } catch(DuplicateKeyException wx) {
            //多表执行必然会出现主键重复，故不入理
            //比离线同步完成时间早的异常不处理,可能引起主键冲突
            //if(checkInOfflineTime(destination, message)) return;
            //log.error("执行SQL异常:", wx);
            //errorSqlDao.writeErrorSql(destination, message, sqlObject,wx.getMessage());
        } catch(RecoverableDataAccessException cx) { //连接问题
            log.warn("出现连接异常:", cx);
            execRunSql(todb, message, sqlObject);
        } catch (Exception ex) {
            log.error("执行SQL异常:", ex);
            errorSqlDao.writeErrorSql(todb.getKey(), message, sqlObject,ex.getMessage());
        }
        HandlerDataSource.clear();
    }
    //false 记录,true 不记录
    /*private boolean checkInOfflineTime(String destination, DtsBinlogMessage message) {
        SynObjectKey key = createKey(destination, message);
        OfflineStateInfo offlineStateInfo = offlineStateInfoDao.get(key);
        if(offlineStateInfo==null) return false;
        if(offlineStateInfo.getState()==null) return false;
        if(OfflineState.NO_START.getState()==offlineStateInfo.getState()) return true;
        if(OfflineState.STARTED.getState()==offlineStateInfo.getState()) return true;
        if(offlineStateInfo.getUpdateTime()==null) return true;
        if(message.getTs()==null) return false;
        if(DateUtil.parseDate(offlineStateInfo.getUpdateTime()).getTime()>new Date(message.getTs()).getTime()) return true;
        return false;
    }*/
    
    /**
     * @param destination
     * @param message
     * @return
     */
    private SynObjectKey createKey(String destination, DtsBinlogMessage message) {
        SynObjectKey key = new SynObjectKey();
        key.setDestination(destination);
        key.setDataSource(message.getDestination());
        key.setDbName(message.getDatabase());
        key.setTableName(message.getTable());
        return key;
    }

    public void execSql(DtsDataBase db,DtsBinlogMessage message) {
        if (message == null) return;
        if (TypeUtil.isNull(message.getExecSqlList())) return;
        if(db.getIsSource()) return;
        for (SqlObject sqlObject : message.getExecSqlList()) {
            execRunSql(db, message, sqlObject);
        }
    }

    private boolean offlineFilter(String destination, DtsBinlogMessage message, SqlObject sqlObject) {
        SynObjectKey key = createKey(destination, message);
        if (!offlineLifeCycle.onRunOffline(key)) return false;
        OfflineSqlTable offlineSqlTable = new OfflineSqlTable();
        BeanUtil.copyProperties(offlineSqlTable, key);
        offlineSqlTable.setPkKey(sqlObject.getPk());
        offlineSqlTable.setSqlStr(sqlObject.getSql());
        offlineSqlTable.setCreateTime(DateUtil.formatDate(new Date()));
        offlineSqlTableDao.save(offlineSqlTable);
        return true;
    }

}
